package com.winning.pmph.utils;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;

import java.util.Objects;
@Slf4j
public class RocketMQProducerUtil {
    private static DefaultMQProducer defaultMQProducer = new DefaultMQProducer("pmph");
    static {
        // 声明并初始化一个producer
        // 需要一个producer group名字作为构造方法的参数，这里为pmph
//        defaultMQProducer = new DefaultMQProducer("pmph");
        defaultMQProducer.setVipChannelEnabled(false);
        // 设置NameServer地址，多个地址之间用；分隔
        defaultMQProducer.setNamesrvAddr(ApplicationContextUtil.getProperty("mq.addr"));
        defaultMQProducer.setDefaultTopicQueueNums(1);
        //消息发送失败重试次数
        defaultMQProducer.setRetryTimesWhenSendFailed(3);
        //异步发送失败重试次数
        defaultMQProducer.setRetryTimesWhenSendAsyncFailed(3);
        // 调用start()方法启动一个producer实例
        try {
            defaultMQProducer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }
    public static DefaultMQProducer getInstance(){
        return defaultMQProducer;
    }

    public static boolean sendMessage(String topic, String tags, String keys, String body) {
        log.info("同步消息:topic:" + topic + ", tags:" + tags + ", keys:" + keys + ", body:" + body);
        try {
            // 封装消息
            Message msg = new Message(topic, tags, keys, body.getBytes(RemotingHelper.DEFAULT_CHARSET));
            // 这里调用的是同步的方式，所以会有返回结果
            SendResult sendResult = defaultMQProducer.send(msg);
            if (Objects.equals(SendStatus.SEND_OK, sendResult.getSendStatus())) {
                log.info("推送结果: " + sendResult);
                return true;
            }else {
                log.error("推送结果: " + sendResult);
                return false;
            }
        } catch (Exception e) {
            log.error("推送异常:topic:" + topic + ", tags:" + tags + ", keys:" + keys + ", body:" + body, e);
            return false;
        }
    }
}
